package cn._51doit.flink.day12;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 从Kafka中读取json数据，然后oid相同的数据覆盖写入到mysql中
 */
public class SQLKafkaToMySQLDemo2 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        env.enableCheckpointing(50000);


        //使用KafkaConnector从Kafka中读取数据
        tEnv.executeSql(
                "CREATE TABLE tb_user_order (\n" +
                        "  `oid` STRING,  \n" +
                        "  `cid` STRING,  \n" +
                        "  `money` DOUBLE \n" +
                        ") WITH (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'order-json',\n" +
                        "  'properties.bootstrap.servers' = 'node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092',\n" +
                        "  'properties.group.id' = 'testGroup',\n" +
                        "  'scan.startup.mode' = 'earliest-offset',\n" +
                        "  'format' = 'json', \n" +
                        "  'json.ignore-parse-errors' = 'true'" +
                        ")"
        );


        //使用sql方式定义sink
        tEnv.executeSql(
                "CREATE TABLE tb_mysql_sink (\n" +
                        "  oid STRING,  \n" +
                        "  cid STRING,  \n" +
                        "  money DOUBLE, \n" +
                        "  PRIMARY KEY (oid) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "   'connector' = 'jdbc',\n" +
                        "   'url' = 'jdbc:mysql://node-1.51doit.cn:3306/doit28',\n" +
                        "   'table-name' = 'tb_order2',\n" +
                        "   'username' = 'root',\n" +
                        "   'password' = '123456'\n" +
                        ")"
        );
        //将Source表和sink表关联到一起
        tEnv.executeSql("INSERT INTO tb_mysql_sink SELECT * FROM tb_user_order where oid is not null");

    }
}
